package adapter

import (
	"encoding/json"
	"fmt"
	"github.com/garyburd/redigo/redis"
	"github.com/spf13/viper"
	"pqstat/pkg/log"
	"time"
)

type RedisQueue struct {
	pool *redis.Pool
}

var pool *redis.Pool

func init() {
	viper.AutomaticEnv()
	viper.SetDefault("PQ_REDIS_HOST", "127.0.0.1")
	viper.SetDefault("PQ_REDIS_PORT", "6379")
	viper.SetDefault("PQ_REDIS_PASSWORD", "")
	host := viper.GetString("PQ_REDIS_HOST")
	port := viper.GetString("PQ_REDIS_PORT")
	pwd := viper.GetString("PQ_REDIS_PASSWORD")
	pool = &redis.Pool{
		MaxIdle:     5,
		MaxActive:   20,
		IdleTimeout: 100 * time.Second,
		Wait:        false,
		Dial: func() (redis.Conn, error) {
			c, err := redis.Dial("tcp", host+":"+port)
			if err != nil {
				log.Fatal(err.Error())
				return nil, err
			}
			if pwd != "" {
				if _, err := c.Do("AUTH", pwd); err != nil {
					log.Fatal(err.Error())
					c.Close()
					return nil, err
				}
			}
			// 选择db
			c.Do("SELECT", 0)
			log.Info("Init one redis connection")
			return c, nil
		},
	}
}
func (rq *RedisQueue) Subscribe(channel string) {}

func (rq *RedisQueue) Consume(channel string, cbs []func(msg interface{})) error {
	sub := redis.PubSubConn{Conn: pool.Get()}
	sub.PSubscribe(channel)
	log.Info(fmt.Sprintf("Redis subscribe start %s", channel))
	for {
		switch v := sub.Receive().(type) {
		case redis.PMessage:
			var data map[string]interface{}
			json.Unmarshal(v.Data, &data)
			//log.Info("%v", data)
			for _, cb := range cbs {
				go cb(data)
			}
		case redis.Message:
			//log.Info(string(v.Data))
		case error:
			return v
		}
	}
	return nil
}
